package com.fanxuankai.boot.mqbroker.consumer.autoconfigure;

import com.alibaba.fastjson.JSON;
import com.fanxuankai.boot.mqbroker.Event;
import com.fanxuankai.boot.mqbroker.autoconfigure.MqBrokerProperties;
import com.fanxuankai.boot.mqbroker.consumer.AbstractMqConsumer;
import com.fanxuankai.boot.mqbroker.consumer.EventListenerRegistry;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;

/**
 * @author fanxuankai
 */
@ConditionalOnClass({RocketMQTemplate.class})
@ConditionalOnProperty(name = MqBrokerConsumerProperties.ENABLED, matchIfMissing = true)
public class MqBrokerConsumerRocketAutoConfiguration {
    private static final Logger LOGGER = LoggerFactory.getLogger(MqBrokerConsumerRocketAutoConfiguration.class);

    @Bean(initMethod = "start")
    public DefaultMQPushConsumer pushConsumer(RocketMQProperties properties,
                                              AbstractMqConsumer<Event<?>> mqConsumer,
                                              EventListenerRegistry eventListenerRegistry) {
        DefaultMQPushConsumer consumer =
                new DefaultMQPushConsumer(properties.getProducer().getGroup());
        consumer.setNamesrvAddr(properties.getNameServer());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            list.forEach(messageExt -> mqConsumer.accept(JSON.parseObject(new String(messageExt.getBody()),
                    Event.class)));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        eventListenerRegistry.getTopics()
                .forEach(s -> {
                    try {
                        consumer.unsubscribe(s);
                        consumer.subscribe(s, "*");
                    } catch (MQClientException e) {
                        LOGGER.error("订阅失败", e);
                    }
                });
        return consumer;
    }

    @Bean
    @ConditionalOnMissingBean
    public AbstractMqConsumer<Event<String>> mqConsumer(MqBrokerProperties properties) {
        return new AbstractMqConsumer<Event<String>>(properties) {
        };
    }
}
